package kafka.client.producer;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * A pool to manage the message sender instances.
 * @author zhaoxiang
 * 2015-07-10
 */
public class MessageSenderPool {

	private static final Logger LOGGER = LoggerFactory.getLogger(MessageSenderPool.class);
	
	private final Semaphore freeSender;
	private final LinkedBlockingQueue<MessageSender> queue;
	private final String topic;
	private final int poolSize;
	private final ExecutorService pool;
	private final Properties props;
	private ReadWriteLock closingLock;
		
	/**
	 * 
	 * A pool to manage the sender instances.
	 * @param size the pool size
	 * @param topic the topic of Kafka
	 * @param brokerStr broker list in String
	 * @param props the config properties
	 */
	public MessageSenderPool(int size, String topic,Properties props)
	{
		closingLock = new ReentrantReadWriteLock();
		poolSize = size;
		this.freeSender = new Semaphore(poolSize);
		this.queue = new LinkedBlockingQueue<MessageSender>(poolSize);
		this.topic = topic;
		pool = Executors.newFixedThreadPool(poolSize);
		this.props = props;
		init();
	}

	
	private void init() 
	{
		List<Callable<Boolean>> taskList = new ArrayList<Callable<Boolean>>();
		final CountDownLatch count = new CountDownLatch(poolSize);
		for(int i=0;i<poolSize;i++)
		{
			taskList.add(new InitTask(count, this));
		}
			
		try {
			pool.invokeAll(taskList);
			count.await(KafkaConstants.INIT_TIMEOUT_MIN, TimeUnit.MINUTES);
		} catch (InterruptedException e) {
			LOGGER.error("Failed to init the MessageSenderPool", e);
		}
	}
	
	class InitTask implements Callable<Boolean>
	{
		CountDownLatch count;
		MessageSenderPool pool;
		
		public InitTask(CountDownLatch count, MessageSenderPool pool) {
			this.count = count;
			this.pool = pool;
		}
		@Override
		public Boolean call() throws Exception {
			MessageSenderImpl sender = new MessageSenderImpl(props, topic, pool);
			if(sender!=null)
			{
				queue.offer(sender);
				count.countDown();
			}
			return true;
		}
	}

	
	/**
	 * Get a sender from the pool within the given timeout
	 * @param waitTimeMillis
	 * how long should it wait for getting the sender instance
	 * @return
	 * a sender instance
	 */
	public MessageSender getSender(long waitTimeMillis) {
		try {
			//获得锁
            if (!freeSender.tryAcquire(waitTimeMillis, TimeUnit.MILLISECONDS)){
            	throw new RuntimeException("Timeout waiting for idle object in the pool.");
            }
            closingLock.readLock().lock();
//    		MessageSender  sender = queue.take();
    		MessageSender sender = queue.poll();
    		if(sender==null)
    		{
    			//获得了锁但是没取到sender,说明池中的sender数量减少了，需新建补充
    			sender = new MessageSenderImpl(props, topic, this);
    			if(sender!=null)
    			{
    				LOGGER.warn("Add new sender to the pool.");
    				queue.offer(sender);
    			}
    		}
    		closingLock.readLock().unlock();
    		return sender;
    		
        } catch (InterruptedException e) {
            throw new RuntimeException("Interrupted waiting for idle object in the pool .");
        }
	}
	
	/**
	 * Return a sender back to pool.
	 */
	public void returnSender(MessageSender sender) {
//		if(this.queue.contains(sender))
//			return;
		this.queue.offer(sender);
		this.freeSender.release();
	}
	
	public synchronized void close() {
		try {
			// lock the thread for closing
			closingLock.writeLock().lock();
			//wait for all sender Return back to pool.
			Thread.sleep(1000);
			List<Callable<Boolean>> taskList = new ArrayList<Callable<Boolean>>();
			int size = queue.size();
			final CountDownLatch count = new CountDownLatch(size);
			for(int i=0;i<size;i++)
			{
				taskList.add(new Callable<Boolean>() {
					@Override
					public Boolean call() throws Exception {
						MessageSender sender = queue.poll();
						if(sender!=null)
						{
							sender.shutDown();
							count.countDown();
						}
						return true;
					}
				});
			}
		
			pool.invokeAll(taskList);
			count.await(KafkaConstants.CLOSE_TIMEOUT_MIN, TimeUnit.MINUTES);
			pool.shutdownNow();
		} catch (InterruptedException e) {
			LOGGER.error("Failed to close the MessageSenderPool", e);
		}
		closingLock.writeLock().unlock();
	}
}
